package dataStream;

import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;
import java.util.List;

/**
 * @desc
 * @Author xuyouchang
 * @Date 2021/10/9
 */
public class KeyByFuncTest {



    public static void main(String[] args) throws Exception {
        List<Person> data = new ArrayList<>();
        data.add( new Person("man", 20));
        data.add( new Person("man", 18));
        data.add( new Person("man", 16));
        data.add( new Person("woman", 20));
        data.add( new Person("woman", 18));
        data.add( new Person("woman", 16));
        data.add( new Person("woman", 15));

        final StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<Person> dataStream = env.fromCollection(data);

//        max 函数
        dataStream.keyBy(person -> person.gender).max("age");

//        reduce 函数
        DataStream<Person> result = dataStream.keyBy("gender").reduce(new ReduceFunction<Person>() {
            @Override
            public Person reduce(Person value1, Person value2) throws Exception {
                return new Person(value1.getGender(), value1.getAge() * value2.getAge());
            }
        });

        result.print();
        env.execute();
    }
}
